import os
import platform
import re
from concurrent.futures import ProcessPoolExecutor, as_completed

import numpy as np
import pandas as pd
from typing import *

from config import PREMIUM, ROOT_DIR
from utils.dataArrange.finalReport import FinalReport
from utils.error import BaseError
from utils.strategy.conf.BtConf import BtConf
from loguru import logger

signal_long = {
    1: 1,  # 单向开多
    0: 0,  # 平
    -1: 0,  # 单向开空
    'BL': 1,  # 开多
    'SL': 0  # 平多
}

signal_short = {
    1: 0,  # 单向开多
    0: 0,  # 平
    -1: 1,  # 单向开空
    'BS': 1,  # 开空
    'SS': 0  # 平平空
}


class MultiExecutor:
    """

    """

    def __init__(self):
        self.report: Optional[FinalReport] = None
        self.multi_cal: bool = False
        self.conf: Optional[BtConf] = None

    @staticmethod
    def _cal_pos(df: pd.DataFrame, signal_name: str = ''):
        """
        单个信号持仓计算
        :param df:
        :param signal_name: 信号列名
        :return:
        """
        signal_col = 'signal' + signal_name

        # 去除连续相同信号
        df_filter = df[signal_col].shift(1).ffill(inplace=True) == df[signal_col]
        df.loc[df_filter, signal_col] = np.nan

        df['_pos_long'] += df[signal_col].map(signal_long).shift(1).ffill().fillna(0)
        df['_pos_short'] += df[signal_col].map(signal_short).shift(1).ffill().fillna(0)

    def calculate(self, df: pd.DataFrame):
        """
        检查所有信号，计算所有信号的回测
        :param df:
        :return:
        """
        df = self.conf.strategy.pick(df)
        if len(self.conf.strategy.symbols) != 1:
            raise BaseError('多信号计算模式只支持单币种')

        signals: List[str] = []
        for col in df.columns:
            res = re.match(r'^signal(\w*)$', col)
            if res is not None:
                signals.append(res.groups()[0])

        if len(signals) == 0:
            raise BaseError(f'未检测到信号数据')

        # 多仓量
        df['_pos_long'] = 0
        # 空仓量
        df['_pos_short'] = 0
        # 余量
        df['_pos_funding'] = 1
        # 仓位成本价格
        df['_long_price'] = np.nan
        df['_short_price'] = np.nan

        # 计算每一个信号
        for item in signals:
            self._cal_pos(df, signal_name=item)

        df['_pos_short'] /= len(signals)
        df['_pos_long'] /= len(signals)
        df['_pos_funding'] -= df['_pos_short'] + df['_pos_long']
        # === 计算当前仓位成本价格
        # 多仓变化
        df['_long_delta'] = df['_pos_long'] - df['_pos_long'].shift(1).fillna(0)
        # 开仓筛选
        open_filter = (df['_long_delta'] > 0) & (df['_pos_long'].shift(1) == 0)
        # 开仓时使用上一个close价作为成本价格
        df.loc[open_filter, '_long_price'] = df['close'].shift(1)[open_filter]
        df['_long_price'].ffill(inplace=True)
        df.loc[df['_pos_long'] == 0, '_long_price'] = np.nan
        # 加仓筛选
        add_filter = (df['_long_delta'] > 0) & (df['_pos_long'].shift(1) != 0)
        # 加仓修改成本价格
        df.loc[add_filter, '_long_price'] = ((df['_long_price'] * df['_pos_long']).shift(1)[add_filter] +
                                             df[add_filter]['close'] * df[add_filter]['_long_delta']) / df[add_filter][
                                                '_pos_long']
        # 重置非加开仓的成本价，然后f填充
        df.loc[df['_long_delta'] <= 0, '_long_price'] = np.nan
        df['_long_price'].ffill(inplace=True)

        # 空仓变化
        df['_short_delta'] = df['_pos_short'] - df['_pos_short'].shift(1).fillna(0)
        # 开仓筛选
        open_filter = (df['_short_delta'] > 0) & (df['_pos_short'].shift(1) == 0)
        # 开仓时使用上一个close价作为成本价格
        df.loc[open_filter, '_short_price'] = df['close'].shift(1)[open_filter]
        df['_short_price'].ffill(inplace=True)
        df.loc[df['_pos_short'] == 0, '_short_price'] = np.nan
        # 加仓筛选
        add_filter = (df['_short_delta'] > 0) & (df['_pos_short'].shift(1) != 0)
        # 加仓修改成本价格
        df.loc[add_filter, '_short_price'] \
            = ((df['_short_price'] * df['_pos_short']).shift(1)[add_filter] +
               df[add_filter]['close'] * df[add_filter]['_short_delta']) / df[add_filter]['_pos_short']
        # 重置非加开仓的成本价，然后f填充
        df.loc[df['_short_delta'] <= 0, '_short_price'] = np.nan
        df['_short_price'].ffill(inplace=True)

        # ==== 计算整体资金变化
        leverage: int = self.conf.strategy.leverage
        premium: float = float(PREMIUM['swap']['taker'])

        df['_income'] = 0
        df['_funding'] = 1
        df['_long_income'] = 0
        df['_short_income'] = 0

        # 手续费
        df['_premium'] = -premium * (df['_long_delta'].abs() + df['_short_delta'].abs()) * leverage
        # 收益率 持仓比例*（收盘价 - 开仓价）* 方向 * 杠杆
        df['_long_income'] = df['_long_delta'] * (df['close'].shift(1) / df['_long_price'] - 1) * leverage
        df['_short_income'] = df['_short_delta'] * (df['_short_price'] / df['close'].shift(1) - 1) * leverage
        # 非减仓的收益置零
        df.loc[df['_long_delta'] >= 0, '_long_income',] = 0
        df.loc[df['_short_delta'] >= 0, '_short_income',] = 0

        # 做空爆仓 价格上升比例*杠杆大于1
        short_liquidation = (df['high'] / df['_short_price'] - 1) * leverage + float(PREMIUM['force']) > 1
        # 做多爆仓 价格下降比例*杠杆小于-1
        long_liquidation = (df['low'] / df['_long_price'] - 1) * leverage + float(PREMIUM['force']) < -1

        # 爆仓替换收益
        df.loc[short_liquidation, '_short_income'] = -df[short_liquidation]['_pos_short']
        df.loc[long_liquidation, '_long_income'] = -df[long_liquidation]['_pos_long']

        df['_funding'] += df['_premium'] + df['_long_income'] + df['_short_income']

        df['_funding'] = df['_funding'].apply(lambda x: x if x >= 0 else 0).cumprod()
        df['_funding_unrealize'] = ''
        return df

    @staticmethod
    def _subprocess(exe: "MultiExecutor", conf: BtConf):
        """
        单次计算
        :param conf:
        :return:
        """
        df = conf.load_data()
        df = df[df['trade_type'] == 1]
        conf.strategy.injection({'factors': conf.strategy.factorFunc})
        exe.conf = conf
        df: pd.DataFrame = exe.calculate(df)
        if exe.multi_cal:
            exe.multi(df, conf)
        else:
            exe.single(df, conf)

    def single(self, df: pd.DataFrame, conf: BtConf):
        """
        单个计算额外执行逻辑
        :param df:
        :param conf:
        :return:
        """
        self.report = FinalReport('report-' + conf.strategy.arg_id)
        self.report.infos.update()
        # 时间字符串
        times: List[str] = pd.to_datetime(df['open_time'], unit='ms').dt.strftime('%Y-%m-%d %H:%M:%S').tolist()
        # 资金线
        funding: List[float] = df['_funding'].round(4).tolist()
        # 资金曲线图
        self.report.account_funding(times, 'funding', funding, df['close'].tolist())
        # 信号点位图
        self.report.cta_infos(times, funding, df, factors=conf.show_factors)
        self.report.generate(os.path.join(ROOT_DIR, 'locals', 'output'))

        if platform.system() == 'Windows':
            html_f_name = f'report-{conf.strategy.arg_id}' + '.html'
            os.system(fr'start {os.path.join(ROOT_DIR, "locals/output", html_f_name)}')

    @staticmethod
    def multi(df: pd.DataFrame, conf: BtConf):
        """
        多任务计算额外执行逻辑
        :param df:
        :param conf:
        :return:
        """
        pass

    def run(self, *confs: BtConf):
        if len(confs) == 0:
            raise BaseError(f'未输入配置')
        elif len(confs) == 1:
            self._subprocess(self, confs[0])
            return

        confs[0].load_data()
        self.multi_cal = True
        with ProcessPoolExecutor(max_workers=min(len(confs), os.cpu_count())) as executor:
            resp = [executor.submit(MultiExecutor._subprocess, self, conf) for conf in confs]
            for i, future in enumerate(as_completed(resp)):
                try:
                    res: Tuple[str, Dict[str, Any]] = future.result()
                except Exception as e:
                    # raise e
                    logger.error(f'子配置执行错误：{confs[i]}，{e.__str__()}')
